package ar.edu.itba.pod.legajo45129.event;

import java.rmi.RemoteException;
import java.rmi.registry.LocateRegistry;
import java.rmi.registry.Registry;
import java.rmi.server.UnicastRemoteObject;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

import org.apache.log4j.Logger;

import ar.edu.itba.event.EventInformation;
import ar.edu.itba.event.RemoteEventDispatcher;
import ar.edu.itba.node.Node;
import ar.edu.itba.node.NodeInformation;
import ar.edu.itba.pod.agent.runner.Agent;
import ar.edu.itba.pod.legajo45129.agentSimulation.DistributedMultithreadEventDispatcher;
import ar.edu.itba.pod.legajo45129.manager.NodeManager;
import ar.edu.itba.pod.multithread.EventDispatcher;

import com.google.common.base.Preconditions;

public class RemoteEventDispatcherImpl implements RemoteEventDispatcher {
	private static Logger logger = Logger.getLogger(RemoteEventDispatcherImpl.class);
	private BlockingQueue<EventInformation> events = new LinkedBlockingQueue<EventInformation>();
	private static final long POLL_INTERVAL = 2000;
	private EventDispatcher eventDispatcher;
	private PublishEvents publishEventsThread;
	
	public RemoteEventDispatcherImpl(EventDispatcher eventDispatcher) throws RemoteException {
		Preconditions.checkNotNull(eventDispatcher);
		UnicastRemoteObject.exportObject(this, 0);
		NodeInformation localNode = NodeManager.getInstance().getNodeInformation();
		Registry registry;
		registry = LocateRegistry.getRegistry(localNode.host(), localNode.port());
		if (registry == null) {
			registry = LocateRegistry.createRegistry(localNode.port());
		}

		logger.debug("Binding " + Node.DISTRIBUTED_EVENT_DISPATCHER);
		registry.rebind(Node.DISTRIBUTED_EVENT_DISPATCHER, this);
		this.eventDispatcher = eventDispatcher;
		this.publishEventsThread = new PublishEvents();
		this.publishEventsThread.start();
	}

	@Override
	public boolean publish(EventInformation event) throws RemoteException,
			InterruptedException {
		if (events.contains(event)) {
			return false;
		}
		events.put(event);
		return true;
	}

	@Override
	public Set<EventInformation> newEventsFor(NodeInformation nodeInformation)
			throws RemoteException {
		synchronized (events) {
			Set<EventInformation> result = new HashSet<EventInformation>();
			for (EventInformation e: events) {
				if (e.nodeId().equals(nodeInformation.id())) {
					result.add(e);
				}
			}
			
			return result;
		}
	}

	@Override
	public BlockingQueue<Object> moveQueueFor(Agent agent)
			throws RemoteException {
		return ((DistributedMultithreadEventDispatcher)eventDispatcher).deregister(agent);
	}
	
	private class PublishEvents extends Thread {
		@Override
		public void run() {
			while (true) {
				try {
					sleep(POLL_INTERVAL);
					EventInformation eventInformation = events.take();
					logger.trace(eventInformation + " catched from queue");
					((DistributedMultithreadEventDispatcher)eventDispatcher).localPublish(eventInformation.source(), eventInformation.event());
				} catch (InterruptedException e) {
					logger.warn("Polling events to put into local event dispatcher Interrupted");
				}
			}
		}
	}

}
